package cn.panshi.mongocdc;

import java.util.Date;
import java.util.UUID;

import org.bson.BsonDocument;
import org.bson.BsonType;
import org.bson.Document;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.messaging.Message;
import org.springframework.data.redis.core.RedisTemplate;

import com.mongodb.client.model.changestream.ChangeStreamDocument;
import com.mongodb.client.model.changestream.OperationType;

import cn.panshi.mongocdc.redis.AbstractRedisLockTemplate;
import cn.panshi.mongocdc.redis.IRedisNxLockProtected;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@SuppressWarnings("rawtypes")
public abstract class AbsCdcTemplate {
	private RedisTemplate redisTemplate;
	private MongoTemplate mongoTemplate;

	public AbsCdcTemplate(RedisTemplate redisTemplate, MongoTemplate mongoTemplate) {
		super();
		this.redisTemplate = redisTemplate;
		this.mongoTemplate = mongoTemplate;
	}

	public final void process(String collectionName, OperationType operationType,
			Message<ChangeStreamDocument<Document>, Document> message) {

		if (!this.supportsOperation(collectionName, operationType)) {
			return;
		}

		String resumeToken = message.getRaw().getResumeToken().getString("_data").getValue();

		String bizId = fetchBizIdFromMessage(message);

		//借助redis分布式锁，避免cdc重复处理
		AbstractRedisLockTemplate redisLock = new AbstractRedisLockTemplate(redisTemplate, 60L,
				"lockToken-" + UUID.randomUUID().toString(), false) {

			@Override
			protected String constructKey() {
				return "mongo_cdc:" + collectionName + "_" + operationType.name() + ":" + resumeToken;
			}
		};

		redisLock.doBizWithinLockProtection(new IRedisNxLockProtected() {

			@Override
			public void onLockFailure(String key) {
				//log.debug("###没有抢到Cdc任务——LockFailure with key {}", key);
			}

			@Override
			public void afterReleaseLock(Long unlock) {
				log.debug("release lock " + unlock);
			}

			@Override
			public void doBizOnLockSuccess(String key) {

				CdcOptLogDoc optLog = saveCdcOptLog(resumeToken, collectionName, operationType);//保存resumeToken，用于重启时选择恢复点
				try {

					executeCdcBiz(bizId, resumeToken, message, collectionName, operationType);

					updateCdcOptLog(optLog, null);
				} catch (Exception e) {
					log.error(e.getMessage(), e);
					updateCdcOptLog(optLog, e);
				}
			}

		});

	}

	abstract protected boolean supportsOperation(String collectionName, OperationType operationType);

	abstract protected void executeCdcBiz(String bizId, String resumeToken,
			Message<ChangeStreamDocument<Document>, Document> message, String collectionName,
			OperationType operationType) throws Exception;

	private void updateCdcOptLog(CdcOptLogDoc optLog, Exception exception) {
		if (exception != null) {
			optLog.setError(ReflectionUtils.calcInvokeStackTrace(exception));
		}

		//固定集合(capped参数为true)的数据不能修改或删除
		//			mongoTemplate.remove(new Query(Criteria.where("_id").is(optLog.getId())), CdcOptLogDoc.TABLE);

		optLog.setCrt(null);
		optLog.setLastUpdate(new Date());
		optLog.setId(UUID.randomUUID().toString());
		mongoTemplate.save(optLog, CdcOptLogDoc.TABLE);
	}

	private CdcOptLogDoc saveCdcOptLog(String resumeToken, String collectionName, OperationType operationType) {
		CdcOptLogDoc doc = new CdcOptLogDoc();
		doc.setId(UUID.randomUUID().toString());
		doc.setCrt(new Date());
		doc.setResumeToken(resumeToken);
		doc.setCollectionName(collectionName);
		doc.setOperationType(operationType.name());

		mongoTemplate.save(doc, CdcOptLogDoc.TABLE);
		return doc;
	}

	public final static String fetchBizIdFromMessage(Message<ChangeStreamDocument<Document>, Document> message) {

		BsonDocument documentKey = message.getRaw().getDocumentKey();
		if (documentKey == null) {
			return null;
		}
		if (documentKey.get("_id") == null) {
			return null;
		}

		if (documentKey.get("_id").getBsonType() == BsonType.STRING) {
			return documentKey.getString("_id").getValue();
		} else {
			if (documentKey.getObjectId("_id").getValue() == null) {
				return null;
			}

			return documentKey.getObjectId("_id").getValue().toString();
		}
	}
}